/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.tabletserver;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.SortedMap;
import java.util.TreeMap;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.client.Cell;
import com.nhncorp.neptune.client.Row;
import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.fs.GPath;
import com.nhncorp.neptune.fs.NeptuneFileSystem;
import com.nhncorp.neptune.tablet.ColumnValue;
import com.nhncorp.neptune.tablet.TabletInfo;
import com.nhncorp.neptune.tabletserver.TabletMapFile.MapFileWriter;

public class ColumnCollection {
  public static final Log LOG = LogFactory.getLog(ColumnCollection.class.getName());
  
  protected NConfiguration conf;
  
  //rowkey, <columnkey, ValueCollection>, ValueCollection = value + timestamp
  protected SortedMap<Row.Key, SortedMap<Cell.Key, ValueCollection>> columnCollectionMap 
          = new TreeMap<Row.Key, SortedMap<Cell.Key, ValueCollection>>();

  public ColumnCollection(NConfiguration conf) {
    this.conf = conf;
  }

  public int addValue(Row.Key rowKey, ColumnValue columnValue, int numOfVersion) throws IOException {
    int saveSize = 0;
    int rowKeyLength = rowKey.getLength();
    int rowKeyAllocLen = rowKey.getAllocatedSize();
    
    SortedMap<Cell.Key, ValueCollection> columnValues = columnCollectionMap.get(rowKey);
    if(columnValues == null) {
      columnValues = new TreeMap<Cell.Key, ValueCollection>();
      columnCollectionMap.put(rowKey, columnValues);
      //saveSize += rowKey.getByteSize();
      saveSize += (rowKeyAllocLen + 80);
    } 
    
    ValueCollection valueCollection = columnValues.get(columnValue.getCellKey());
    if(valueCollection == null) {
      valueCollection = new ValueCollection();
      Cell.Key cellKey = columnValue.getCellKey();
      columnValues.put(cellKey, valueCollection);
      
      saveSize += (144 + ((cellKey.getLength() + 3)/8) * 8); //valueCollection 길이
    }
    
    //saveSize += columnValue.size();
    int duplicatedLength = ((rowKeyLength + 3)/8) * 8;
    saveSize += (columnValue.getAllocatedSize() - duplicatedLength);
    
    valueCollection.add(columnValue, numOfVersion);
    
    return saveSize;
  }

//  public SortedMap<Row.Key, SortedMap<Cell.Key, ValueCollection>> getColumnValues() {
//    return columnValues;
//  }
  
  public void print() {
    for(Map.Entry<Row.Key, SortedMap<Cell.Key, ValueCollection>> entry: columnCollectionMap.entrySet()) {
      Row.Key rowKey = entry.getKey();
      SortedMap<Cell.Key, ValueCollection> columnValues = entry.getValue();
      
      System.out.println("\tRow.Key:" + rowKey);
      for(Map.Entry<Cell.Key, ValueCollection> innerEntry: columnValues.entrySet()) {
        System.out.println("\t\tCell.Key:" + innerEntry.getKey());
        innerEntry.getValue().print();
      }
    }
  }

  public ColumnValue get(Row.Key rowKey, Cell.Key columnKey) {
    SortedMap<Cell.Key, ValueCollection> valueCollections = columnCollectionMap.get(rowKey);
    if(valueCollections == null)  return null;
    
    ValueCollection value = valueCollections.get(columnKey);
    if(value == null) return null;
    return value.get();
  }

  public ValueCollection getValueCollection(Row.Key rowKey, Cell.Key columnKey) {
    SortedMap<Cell.Key, ValueCollection> valueCollections = columnCollectionMap.get(rowKey);
    if(valueCollections == null)  return null;
    
    return valueCollections.get(columnKey);
  }
  
  public ColumnValue[] get(Row.Key rowKey) {
    SortedMap<Cell.Key, ValueCollection> columnValues = columnCollectionMap.get(rowKey);
    if(columnValues == null) {
      return null;
    }
    
    List<ColumnValue> columnValueList = new ArrayList<ColumnValue>(10);
    for(ValueCollection value: columnValues.values()) {
      if(!value.isDeleted()) {
        columnValueList.add(value.get());
      }
    }
    if(columnValueList.size() == 0) {
      return null;
    }
    
    return columnValueList.toArray(new ColumnValue[]{});
  }
  
  public ColumnValue[][] getAllVersion(Row.Key rowKey, Cell.Key columnKey) {
    SortedMap<Cell.Key, ValueCollection> columnValues = columnCollectionMap.get(rowKey);
    //LOG.debug("get:" + rowKey + ":" + columnValues);
    if(columnValues == null) {
      return null;
    }
    
    if(columnKey == null || columnKey.isEmpty()) {
      //모든 columnKey에 대해 조회하는 경우
      ColumnValue[][] result = new ColumnValue[columnValues.size()][];
      int index = 0;
      for(ValueCollection value: columnValues.values()) {
        result[index++] = value.get(Integer.MAX_VALUE);
      }
      if(result.length == 0) {
        return null;
      }
      
      return result;
    } else {
      //특정 columnKey에 대해서 조회하는 경우
      ValueCollection valueCollection = columnValues.get(columnKey);
      if(valueCollection == null) {
        return null;
      } else {
        ColumnValue[][] result = new ColumnValue[1][];
        result[0] = valueCollection.get(Integer.MAX_VALUE);
        return result;
      }
    }
  }
  
  public void print(Row.Key rowKey) {
    SortedMap<Cell.Key, ValueCollection> columnValues = columnCollectionMap.get(rowKey);
    //LOG.debug("get:" + rowKey + ":" + columnValues);
    if(columnValues == null) {
      System.out.println("null");
      return;
    }
    
    for(ValueCollection value: columnValues.values()) {
      value.print();
    }
  }

  /**
   * 데이터를 임시 디렉토리에 저장한다.
   * @param tabletInfo
   * @param columnName
   * @return
   * @throws IOException
   */
  public TabletMapFile saveToDisk(TabletInfo tabletInfo, 
      String columnName, 
      String fileId,
      int numOfVersion) throws IOException {
    //데이터가 없으면 저장하지 않는다.
    if(columnCollectionMap.size() == 0) {
      return null;
    }
    
    GPath tempMapFilePath = 
      new GPath(Tablet.getTabletMinorCompactionTempPath(conf, tabletInfo), 
          columnName + "/" + fileId + "/");
    
    return saveToDisk(NeptuneFileSystem.get(conf),
        tabletInfo, 
        columnName, 
        tempMapFilePath, 
        fileId,
        numOfVersion);
  }

  /**
   * BatchUpload를 위해 임시로 로컬 파일시스템에 저장시킨다.
   * @param fs
   * @param tabletInfo
   * @param columnName
   * @param fileId
   * @return
   * @throws IOException
   */
  public TabletMapFile saveToLocalDisk(NeptuneFileSystem fs, 
      TabletInfo tabletInfo, 
      String columnName, 
      String fileId,
      String uploadAcitonId,
      int numOfVersion) throws IOException {
    if(columnCollectionMap.size() == 0){
      return null;
    }
    GPath uploadMapFilePath = new GPath(Tablet.getTabletLocalTempPath(conf, tabletInfo, uploadAcitonId), columnName + "/" + fileId + "/");
    return saveToDisk(fs, tabletInfo, columnName, uploadMapFilePath, fileId, numOfVersion);
  }
  
  private TabletMapFile saveToDisk(NeptuneFileSystem fs, TabletInfo tabletInfo, 
      String columnName, GPath parentPath, String fileId, int numOfVersion) throws IOException {
    TabletMapFile tabletMapFile = new TabletMapFile(conf, fs, tabletInfo, columnName, fileId, parentPath, numOfVersion);
    MapFileWriter mapFileWriter = tabletMapFile.getMapFileWriter();

    try {
      for(Map.Entry<Row.Key, SortedMap<Cell.Key, ValueCollection>> entry: columnCollectionMap.entrySet()) {
        for(Map.Entry<Cell.Key, ValueCollection> eachEntry: entry.getValue().entrySet()) {
          mapFileWriter.write(eachEntry.getValue());
        }
      }
    } finally {
      mapFileWriter.close();
    }
    //LOG.info("saveToDisk:writeCount=" + writeCount);
    return tabletMapFile;
  }

  public TabletMapFile[] splitAndSaveToDisk( TabletInfo tabletInfo, Row.Key midRowKey, 
                                              TabletInfo[] splitedTabletInfos, 
                                              String columnName, 
                                              String fileId,
                                              int numOfVersion) throws IOException {
    TabletMapFile[] tabletMapFiles = new TabletMapFile[2];
    MapFileWriter[] mapFileWriters = new MapFileWriter[2]; 

    for(int i = 0; i < 2; i++) {
      GPath tempMapFilePath = new GPath(Tablet.getTabletSplitTempPath(conf, splitedTabletInfos[i]), columnName + "/" + fileId + "/");
      tabletMapFiles[i] = new TabletMapFile(conf, 
          NeptuneFileSystem.get(conf),
          splitedTabletInfos[i], columnName, fileId, tempMapFilePath,
          numOfVersion);
      mapFileWriters[i] = tabletMapFiles[i].getMapFileWriter();
    }
    
    int index = 0;
    try {
      for(Map.Entry<Row.Key, SortedMap<Cell.Key, ValueCollection>> entry: columnCollectionMap.entrySet()) {
        Row.Key rowKey = entry.getKey();
        
        if(index < 1 && rowKey.compareTo(midRowKey) > 0) {
          index++;
        }
        for(Map.Entry<Cell.Key, ValueCollection> eachEntry: entry.getValue().entrySet()) {
          mapFileWriters[index].write(eachEntry.getValue());
        }
        
      }
    } finally {
      for(int i = 0; i < 2; i++) {
        try {   
          mapFileWriters[i].close();  
        } catch (IOException e) { 
          LOG.error(e.getMessage(), e); 
        }
      }
    }
    return tabletMapFiles;    
  }
  
  /**
   * MEAT, ROOT tablet인 경우만 사용된다.
   * 하나의 rowkey에 컬럼의 데이터는 하나만 저장되고, 삭제되지 않은 데이터중 가장 근접한 값을 반환한다.
   * @param rowKey
   * @return
   */
  public ColumnValue findNearestValue(Row.Key rowKey) {
    SortedMap<Row.Key, SortedMap<Cell.Key, ValueCollection>> tailMap = columnCollectionMap.tailMap(rowKey);
    //LOG.debug("findNearestRow.Key:rowKey=" + rowKey);
    if(tailMap.size() > 0) {
      for(Map.Entry<Row.Key, SortedMap<Cell.Key, ValueCollection>> entry: tailMap.entrySet()) {
        Row.Key entryRowKey = entry.getKey();
        
        for(ValueCollection eachValueCollection: entry.getValue().values()) {
          //LOG.debug("findNearestRow.Key:valueCollection.get()=" + valueCollection.getWithDeleted());
          //GRID-63
          if(eachValueCollection.get() != null) {
            return eachValueCollection.get();
          }
        }
      }
      return null;
    } else {
      return null;
    }  
  }

  public int getRowCount() {
    return columnCollectionMap.size();
  }
  
  public static void main(String[] args) throws Exception {
    /*
    Assuming 32-bit words and a 12 byte overhead per array instance,
    int[4,100,000][2][2][5]
    4th dim. = overhead + length * sizeof(int) = 12 + 5 * 4 = 32
    3rd dim. = overhead + length * (sizeof(ref) + sizeof(4th_dim)) = 12 + 2 * (4 + 32) = 84
    2nd dim. = overhead + length * (sizeof(ref) + sizeof(3rd_dim)) = 12 + 2 * (4 + 84) = 188
    1st dim. = overhead + length * (sizeof(ref) + sizeof(2nd_dim)) = 12 + 4,100,000 * (4 + 188) = 787,200,012
    
    byte[]의 길이 = 12 + ((length + 3)/8) * 8 + 4
    
    byte의 경우 0 ~ 4까지는 12 + 4가 되지만
     그 이상의 경우 8씩 증가
     예: 0 ~ 4: 16, 5 ~ 12: 24, 13 ~ 20: 32 
     
     따라서 Text의 경우 객체 header 4byte, byte[] 레퍼런스 4byte, int length 4byte, length의 레퍼런스 4 byte = 16 + len(byte[])
    */    

    //Cell.Key, Row.Key: 16byte + byte[]
    //ColumnValue: 16 + rowKey, columnKey, value
    //ColumnCollection: ColumnValue + 124
    
    Runtime runtime = Runtime.getRuntime();
    Cell.Key columnKey = new Cell.Key("123");
    Row.Key rowKey = new Row.Key("123");
    ColumnValue columnValue = new ColumnValue();
        
    ColumnCollection collection = new ColumnCollection(new NConfiguration());
    collection.addValue(rowKey, new ColumnValue(rowKey, columnKey, "".getBytes()), 0);
    
    byte[] valueBuf = "0123456789".getBytes();
    byte[] columnKeyBuf = "abcdefghij".getBytes();
    byte[] rowKeyBuf = "!@#$%^&*()".getBytes();

    Cell.Key columnKey2 = new Cell.Key("qwertyuiop".getBytes());
    Row.Key rowKey2 = new Row.Key("asdfghjkl;".getBytes());
    ColumnValue columnValue2 = new ColumnValue(rowKey2, columnKey2, "zxcvbnm,./".getBytes());

    columnKey = new Cell.Key(columnKeyBuf);
    rowKey = new Row.Key(rowKeyBuf);
    columnValue = new ColumnValue(rowKey, columnKey, valueBuf);
    
    runtime.gc();
    long before = runtime.freeMemory();

    collection.addValue(rowKey, columnValue, 0);
    collection.addValue(rowKey2, columnValue2, 0);
    
    runtime.gc();
    long after = runtime.freeMemory();
    
    int byteLen = 12 + ((valueBuf.length + 3) / 8) * 8 + 4;
    int rowKeyLen = 16 + byteLen;
    int columnKeyLen = 16 + byteLen;
    
    int columnValueLen = 16 + rowKeyLen + columnKeyLen + byteLen;
    
    System.out.printf("%d - %d = %d, %d\n", before, after, (before-after), columnValueLen);
    //24(btey[]) + 40(rowkey) + 40(columnkey)
    
    //static 변수를 위해 임시로 하나 생성
    
    
//    ColumnCollection collection = new ColumnCollection(new NeptuneConfiguration());
//    
//    Row.Key rowKey = new Row.Key("0123456789");
//    Cell.Key columnKey = new Cell.Key("0123456789");
//    ColumnValue columnValue = new ColumnValue(rowKey, columnKey, "0123456789".getBytes());
//    int size = collection.addValue(rowKey, columnValue, 0);
//    
//    SortedMap<Row.Key, SortedMap<Cell.Key, ValueCollection>> columnCollectionMap 
//    = new TreeMap<Row.Key, SortedMap<Cell.Key, ValueCollection>>();
//    
//    int saveSize = 0;
//    SortedMap<Cell.Key, ValueCollection> columnValues = columnCollectionMap.get(rowKey);
//    if(columnValues == null) {
//      columnValues = new TreeMap<Cell.Key, ValueCollection>();
//      columnCollectionMap.put(rowKey, columnValues);
//      saveSize += rowKey.getByteSize();
//    } 
//    
//    ValueCollection valueCollection = columnValues.get(columnValue.getCell.Key());
//    if(valueCollection == null) {
//      valueCollection = new ValueCollection();
//      columnValues.put(columnValue.getCell.Key(), valueCollection);
//    }
//    saveSize += columnValue.size();
//    
//    valueCollection.add(columnValue, 0);
//    
//    System.out.println(">>>>" + size);
    
  }
  
//  우선 몇 가지를 알려드리면,
//  1) 객체 자체가 최소한 4바이트를 차지합니다.
//  아무런 필드가 없는 빈 객체가 4바이트입니다.
//
//  반면에 String 개체는 20바이트를 차지합니다.
//  객체 header 4바이트
//  length 필드 4바이트
//  스트링이 저장된 공간에 대한 레퍼런스 4바이트
//  해쉬값으로 4파이트
//  char[]에 대한 레퍼런스 (또는 포인터) 4 바이트
//
//  char[] 도 객체이므로, 최소한 4바이트를 차지합니다.
//  거기에 실제 데이터 공간이 필요합니다.
//
//  따라서, new String("0123456789") 는 최소한
//  20 + 4 + 10*2 바이트를 차지합니다.
//
//  2) ArrayList에서 String 객체를 가르키는데, 4바이트를 차지합니다.
//  그리고, ArrayList는 크기가 커지면, 1.5 배+ 1개로 그 용량을 늘립니다.
//
//  3) Java에서 new String("abc" 는 데이터를 복제하지 않습니다.
//  즉, new String("01234...789" 는 Sring이라는 객체는 만들지만, 문자열 저장 공간은 공유를 합니다.
//
//  따라서, 2번 프로그램에서 실제 "0..9" 까지의 20글자짜리 문자열은
//  단 한개만 존재합니다.
//  String 객체는 그것에 대한 포인터(레퍼런스) 만을 갖고 있을 뿐입니다.  
}
